package com.dis.fan;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink CDC 2.2.1, 捕获MySQL数据表变化，并写入MySQL
 *
 * @author 闻武
 * @since 2020-05-31
 */
public class CdcMySQL2MySQL {
    public static void main(String[] args) throws Exception {
        /**
         * 1 > 程序入口
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.disableOperatorChaining();

        /**
         * 2> 输入参数，可参考： -hostname 192.168.123.58 -port 3306 -username root -password 123456 -databaseName mydb
         * （Idea环境，Run ---> Edit Configurations，打开Run/Debug Configurations，在相关工程的Program arguments中输入）
         */
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get("hostname");
        int port = parameterTool.getInt("port");
        String username = parameterTool.get("username");
        String password = parameterTool.get("password");
        String databaseName = parameterTool.get("databaseName");

        /**
         * 3> 流表
         */
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String strSourceSql = "CREATE TABLE IF NOT EXISTS products_mys_cdc (\n" +
                "    id INT,\n" +
                "    name STRING,\n" +
                "    description STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = '" + hostname + "',\n" +
                "    'port' = '" + port + "',\n" +
                "    'username' = '" + username + "',\n" +
                "    'password' = '" + password + "',\n" +
                "    'database-name' = '" + databaseName + "',\n" +
                "    'table-name' = 'products',\n" +
                "    'debezium.log.mining.continuous.mine'='true',\n"+
                "    'debezium.log.mining.strategy'='online_catalog',\n" +
                "    'debezium.database.tablename.case.insensitive'='false',\n"+
                "    'scan.startup.mode' = 'initial')";

        String strSinkSql = " CREATE TABLE IF NOT EXISTS products_mys_sink (" +
                "    id INT,\n" +
                "    name STRING,\n" +
                "    description STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'jdbc'," +
                "    'url' = 'jdbc:mysql://" + hostname + ":" + port + "/" + databaseName + "'," +
                "    'table-name' = 'products_sink'," +
                "    'username' = '" + username + "'," +
                "    'password' = '" + password + "' " +
                " )";

        tableEnv.executeSql(strSourceSql);
        tableEnv.executeSql(strSinkSql);
        tableEnv.executeSql("insert into products_mys_sink select * from products_mys_cdc ");
    }
}